package org.xq.softcup.mq.client.producer;

import org.xq.softcup.mq.client.consumer.annotation.MqConsumer;
import org.xq.softcup.mq.client.factory.MqClientFactory;
import org.xq.softcup.mq.client.message.MqMessage;
import org.xq.softcup.mq.client.message.MqMessageStatus;
import org.xq.softcup.mq.client.util.LogUtil;
import org.xq.softcup.mq.util.IpUtil;

import java.util.Date;
import java.util.Set;

/**
 * @author muxin
 */
public class MqProducer {
    /**
     * 发送消息
     * @param mqMessage
     */
    public static void produce(MqMessage mqMessage){
        produce(mqMessage, true);
    }
    /**
     * 发送消息
     * @param mqMessage
     * @param async  是否异步执行
     */
    public static void produce(MqMessage mqMessage, boolean async){
        //检查消息
        validMessage(mqMessage);
        //发送消息
        MqClientFactory.addMessages(mqMessage, async);
    }


    /**
     * 广播异步消息发送
     */
    public static void broadcast(MqMessage mqMessage, boolean async){
        //检查消息
        validMessage(mqMessage);

        //查找在线的组
        Set<String> groupList = MqClientFactory.getConsumerRegistrySupport().getTotalGroupList(mqMessage.getTopic());

        for (String group: groupList) {
            //复制消息
            MqMessage cloneMsg = new MqMessage(mqMessage);
            cloneMsg.setGroup(group);
            //
            produce(cloneMsg, true);
        }
    }

    /**
     * 广播发送消息
     * @param mqMessage
     */
    public static void broadcast(MqMessage mqMessage){
        broadcast(mqMessage, true);
    }


    /**
     * 校验消息
     * @param mqMessage
     * @return
     */
    public static void validMessage(MqMessage mqMessage){
        if (mqMessage == null) {
            throw new IllegalArgumentException("MqMessage不能为空.");
        }
        // 主题
        if (mqMessage.getTopic()==null || mqMessage.getTopic().trim().length()==0) {
            throw new IllegalArgumentException("topic不能为空.");
        }
        if (!(mqMessage.getTopic().length()>=4 && mqMessage.getTopic().length()<=255)) {
            throw new IllegalArgumentException("topic长度在4~255字符之间.");
        }

        // 分组
        if (mqMessage.getGroup()==null || mqMessage.getGroup().trim().length()==0) {
            mqMessage.setGroup(MqConsumer.DEFAULT_GROUP);
        }
        if (!(mqMessage.getGroup().length()>=4 && mqMessage.getGroup().length()<=255)) {
            throw new IllegalArgumentException("group长度在4~255字符之间.");
        }

        // 消息体
        if (mqMessage.getData() == null) {
            mqMessage.setData("");
        }
        if (mqMessage.getData().length() > 20000) {
            throw new IllegalArgumentException("数据长度过长,应在20000字符以内.");
        }

        // 状态
        mqMessage.setStatus(MqMessageStatus.NEW.name());

        // 重试次数
        if (mqMessage.getRetryCount() < 0) {
            mqMessage.setRetryCount(0);
        }

        // 分片ID
        if (mqMessage.getShardingId() < 0) {
            mqMessage.setShardingId(0);
        }

        // 延迟时间
        if (mqMessage.getEffectTime() == null) {
            mqMessage.setEffectTime(new Date());
        }

        // 超时时间
        if (mqMessage.getTimeout() < 0) {
            mqMessage.setTimeout(0);
        }

        // 日志
//        String appendLog = LogUtil.makeLog("生产消息", "消息生产者IP=" + IpUtil.getIp() );
//        mqMessage.setLog(appendLog);
    }
}
